package com.mall.order.mq.promoOrder;

import com.alibaba.fastjson.JSON;
import com.mall.order.OrderPromoService;
import com.mall.order.constant.OrderRetCode;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.order.dto.CreateSeckillOrderResponse;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.Reference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;

/**
 * Author  kkw
 * Date  2021/6/25 15:47
 * Version 1.0
 * Description :
 */
@Component
@Slf4j
public class PromoOrderConsumer {

    @Reference
    OrderPromoService orderPromoService;

    private DefaultMQPushConsumer consumer;

    @Value("${mq.nameserver.address}")
     String nameserverAddress;
    @Value(("${mq.topicname}"))
    private String topicName;


    @PostConstruct
    public void init() {
        consumer = new DefaultMQPushConsumer("promo_order_consumer");
//        consumer.setNamesrvAddr(nameserverAddress);
        consumer.setNamesrvAddr(nameserverAddress);

        try {
            consumer.subscribe(topicName,"*");
            //监听消息,实现消费逻辑,生成订单
            consumer.setMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    MessageExt messageExt = list.get(0);
                    byte[] body = messageExt.getBody();
                    CreateSeckillOrderRequest request = JSON.parseObject(new String(body, 0, body.length, StandardCharsets.UTF_8), CreateSeckillOrderRequest.class);
                    CreateSeckillOrderResponse response = orderPromoService.createPromoOrder(request);
                    if (OrderRetCode.SUCCESS.getCode().equals(response.getCode())) {
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return null;
                }
            });

            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }
}
